#include <Disks/ObjectStorages/HDFS/HDFSObjectStorage.h>

#include <IO/copyData.h>
#include <Storages/HDFS/WriteBufferFromHDFS.h>
#include <Storages/HDFS/HDFSCommon.h>

#include <Storages/HDFS/ReadBufferFromHDFS.h>
#include <Disks/IO/ReadBufferFromRemoteFSGather.h>
#include <Common/getRandomASCIIString.h>


#if USE_HDFS

namespace DB
{

namespace ErrorCodes
{
    extern const int UNSUPPORTED_METHOD;
    extern const int HDFS_ERROR;
}

void HDFSObjectStorage::shutdown()
{
}

void HDFSObjectStorage::startup()
{
}

ObjectStorageKey HDFSObjectStorage::generateObjectKeyForPath(const std::string & /* path */) const
{
    /// what ever data_source_description.description value is, consider that key as relative key
    return ObjectStorageKey::createAsRelative(hdfs_root_path, getRandomASCIIString(32));
}

bool HDFSObjectStorage::exists(const StoredObject & object) const
{
    const auto & path = object.remote_path;
    const size_t begin_of_path = path.find('/', path.find("//") + 2);
    const String remote_fs_object_path = path.substr(begin_of_path);
    return (0 == hdfsExists(hdfs_fs.get(), remote_fs_object_path.c_str()));
}

std::unique_ptr<ReadBufferFromFileBase> HDFSObjectStorage::readObject( /// NOLINT
    const StoredObject & object,
    const ReadSettings & read_settings,
    std::optional<size_t>,
    std::optional<size_t>) const
{
    return std::make_unique<ReadBufferFromHDFS>(object.remote_path, object.remote_path, config, patchSettings(read_settings));
}

std::unique_ptr<ReadBufferFromFileBase> HDFSObjectStorage::readObjects( /// NOLINT
    const StoredObjects & objects,
    const ReadSettings & read_settings,
    std::optional<size_t>,
    std::optional<size_t>) const
{
    auto disk_read_settings = patchSettings(read_settings);
    auto read_buffer_creator =
        [this, disk_read_settings]
        (bool /* restricted_seek */, const StoredObject & object_) -> std::unique_ptr<ReadBufferFromFileBase>
    {
        const auto & path = object_.remote_path;
        size_t begin_of_path = path.find('/', path.find("//") + 2);
        auto hdfs_path = path.substr(begin_of_path);
        auto hdfs_uri = path.substr(0, begin_of_path);

        return std::make_unique<ReadBufferFromHDFS>(
            hdfs_uri, hdfs_path, config, disk_read_settings, /* read_until_position */0, /* use_external_buffer */true);
    };

    return std::make_unique<ReadBufferFromRemoteFSGather>(
        std::move(read_buffer_creator), objects, "hdfs:", disk_read_settings, nullptr, /* use_external_buffer */false);
}

std::unique_ptr<WriteBufferFromFileBase> HDFSObjectStorage::writeObject( /// NOLINT
    const StoredObject & object,
    WriteMode mode,
    std::optional<ObjectAttributes> attributes,
    size_t buf_size,
    const WriteSettings & write_settings)
{
    if (attributes.has_value())
        throw Exception(
            ErrorCodes::UNSUPPORTED_METHOD,
            "HDFS API doesn't support custom attributes/metadata for stored objects");

    /// Single O_WRONLY in libhdfs adds O_TRUNC
    return std::make_unique<WriteBufferFromHDFS>(
        object.remote_path, config, settings->replication, patchSettings(write_settings), buf_size,
        mode == WriteMode::Rewrite ? O_WRONLY : O_WRONLY | O_APPEND);
}


/// Remove file. Throws exception if file doesn't exists or it's a directory.
void HDFSObjectStorage::removeObject(const StoredObject & object)
{
    const auto & path = object.remote_path;
    const size_t begin_of_path = path.find('/', path.find("//") + 2);

    /// Add path from root to file name
    int res = hdfsDelete(hdfs_fs.get(), path.substr(begin_of_path).c_str(), 0);
    if (res == -1)
        throw Exception(ErrorCodes::HDFS_ERROR, "HDFSDelete failed with path: {}", path);

}

void HDFSObjectStorage::removeObjects(const StoredObjects & objects)
{
    for (const auto & object : objects)
        removeObject(object);
}

void HDFSObjectStorage::removeObjectIfExists(const StoredObject & object)
{
    if (exists(object))
        removeObject(object);
}

void HDFSObjectStorage::removeObjectsIfExist(const StoredObjects & objects)
{
    for (const auto & object : objects)
        removeObjectIfExists(object);
}

ObjectMetadata HDFSObjectStorage::getObjectMetadata(const std::string &) const
{
    throw Exception(
        ErrorCodes::UNSUPPORTED_METHOD,
        "HDFS API doesn't support custom attributes/metadata for stored objects");
}

void HDFSObjectStorage::copyObject( /// NOLINT
    const StoredObject & object_from,
    const StoredObject & object_to,
    const ReadSettings & read_settings,
    const WriteSettings & write_settings,
    std::optional<ObjectAttributes> object_to_attributes)
{
    if (object_to_attributes.has_value())
        throw Exception(
            ErrorCodes::UNSUPPORTED_METHOD,
            "HDFS API doesn't support custom attributes/metadata for stored objects");

    auto in = readObject(object_from, read_settings);
    auto out = writeObject(object_to, WriteMode::Rewrite, /* attributes= */ {}, /* buf_size= */ DBMS_DEFAULT_BUFFER_SIZE, write_settings);
    copyData(*in, *out);
    out->finalize();
}


std::unique_ptr<IObjectStorage> HDFSObjectStorage::cloneObjectStorage(const std::string &, const Poco::Util::AbstractConfiguration &, const std::string &, ContextPtr)
{
    throw Exception(ErrorCodes::UNSUPPORTED_METHOD, "HDFS object storage doesn't support cloning");
}

}

#endif
